package com.poetic.network.flow.process;

import com.poetic.network.flow.domain.ItemViewCount;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <pre>
 *  TODO
 * Created by lianghuikun on 2020-09-14.
 * </pre>
 *
 * @author lianghuikun
 */
// 排序
public class TopNHotItems extends KeyedProcessFunction<Long, ItemViewCount, String> {

    private int topSize;

    public TopNHotItems(int topSize) {
        this.topSize = topSize;
    }

    private ListState<ItemViewCount> itemState;

    @Override
    public void open(Configuration parameters) throws Exception {
        itemState = getRuntimeContext().getListState(new ListStateDescriptor("itemState",
                TypeInformation.of(new TypeHint<ItemViewCount>() {
                })));
    }


    @Override
    public void processElement(ItemViewCount value, Context ctx, Collector<String> out) throws Exception {
        // 1.把每条数据存储状态列表
        itemState.add(value);
        // 2.注册定时器(定一个延迟触发的定时器，以统计数据）
        ctx.timerService().registerProcessingTimeTimer(value.getWindowEnd() + 1);

    }

    // 3.定时器触发时，对所有数据（同一窗口下），并输出结果
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        // 4.将所有state中的数据取出放到一个list中
        List<ItemViewCount> itemViewList = new ArrayList<>();
        Iterable<ItemViewCount> iterable = itemState.get();
        iterable.forEach(view -> itemViewList.add(view));
        // 5.按照count大小排序
        List<ItemViewCount> sortedItemList = itemViewList.stream()
                .sorted(Comparator.comparingLong(ItemViewCount::getCount).reversed())
                .limit(topSize)
                .collect(Collectors.toList());
        // 6.清空
        itemState.clear();
        // 7.将排名结果格式化输出

        StringBuilder result = new StringBuilder();
        result.append("时间：").append(new Timestamp(timestamp - 1)).append("\n");
        // 每一个商品的信息
        for (int i = 0; i < sortedItemList.size(); i++) {
            ItemViewCount itemViewCount = sortedItemList.get(i);
            result.append("No").append(i + 1).append(":")
                    .append(" 商品id=").append(itemViewCount.getItemId())
                    .append(" 浏览量=").append(itemViewCount.getCount())
                    .append("\n");

        }
        result.append("====================================\n\n");
        // 控制输出频率，模拟实时滚动结果
        TimeUnit.SECONDS.sleep(1);
        out.collect(result.toString());
    }

}
